package com.fengpb.friday.conductor.core.execution;

import com.fengpb.friday.conductor.common.metadata.run.Workflow;
import static com.fengpb.friday.conductor.common.metadata.run.Workflow.*;
import com.fengpb.friday.conductor.common.metadata.task.Task;
import static com.fengpb.friday.conductor.common.metadata.task.Task.Status;
import static com.fengpb.friday.conductor.common.metadata.task.Task.Status.*;
import com.fengpb.friday.conductor.common.metadata.workflow.WorkflowDef;
import com.fengpb.friday.conductor.common.metadata.workflow.WorkflowTask;
import com.fengpb.friday.conductor.core.execution.mapper.TaskMapper;
import com.fengpb.friday.conductor.core.execution.mapper.TaskMapperContext;
import com.fengpb.friday.conductor.core.utils.IDGenerator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 决策器
 */
@Component
public class DeciderService {

    private static final Logger log = LoggerFactory.getLogger(DeciderService.class);

    private final ParametersUtils parametersUtils;

    private final Map<String, TaskMapper> taskMappers;

    public DeciderService(ParametersUtils parametersUtils, Map<String, TaskMapper> taskMappers) {
        this.parametersUtils = parametersUtils;
        this.taskMappers = taskMappers;
    }

    /**
     * 决策流程
     * @param workflow 流程实例
     * @return 决策结论
     * @throws TerminateWorkflowException 决策异常
     */
    public DeciderOutcome decide(Workflow workflow) throws TerminateWorkflowException {
        final List<Task> tasks = workflow.getTasks();

        List<Task> tasksToBeScheduled = new LinkedList<>();
        if (tasks.isEmpty()) {
            tasksToBeScheduled = startWorkflow(workflow);
            if (tasksToBeScheduled == null) {
                tasksToBeScheduled = new LinkedList<>();
            }
        }
        return decide(workflow, tasksToBeScheduled);
    }

    public DeciderOutcome decide(final Workflow workflow, List<Task> preScheduledTasks) {
        DeciderOutcome outcome = new DeciderOutcome();

        if (workflow.getStatus().isTerminal()) {
            log.debug("Workflow {} is already finished. Reason: {}", workflow, workflow.getReasonForInCompletion());
            return outcome;
        }

        Map<String, Task> tasksToBeScheduled = new LinkedHashMap<>();

        preScheduledTasks.forEach(preScheduledTask -> {
            tasksToBeScheduled.put(preScheduledTask.getReferenceTaskName(), preScheduledTask);
        });

        for (Task pendingTask : preScheduledTasks) {
            List<Task> nextTasks = getNextTask(workflow, pendingTask);
            nextTasks.forEach(nextTask -> tasksToBeScheduled.putIfAbsent(nextTask.getReferenceTaskName(), nextTask));
        }

        List<Task> unScheduledTasks = new ArrayList<>(tasksToBeScheduled.values());
        outcome.tasksToBeScheduled.addAll(unScheduledTasks);

        if (outcome.tasksToBeScheduled.isEmpty() && checkForWorkflowCompletion(workflow)) {
            log.debug("Marking workflow: {} as complete.", workflow);
            outcome.isComplete = true;
        }

        return outcome;
    }

    /**
     * 启动流程
     * @param workflow 流程实例
     * @return 首次启动需执行的任务集合
     */
    public List<Task> startWorkflow(Workflow workflow) {
        final WorkflowDef workflowDef = workflow.getWorkflowDefinition();

        log.debug("Starting workflow: {}", workflow);

        List<Task> tasks = workflow.getTasks();
        if (tasks.isEmpty()) {
            if (workflowDef.getTasks().isEmpty()) {
                throw new TerminateWorkflowException("No tasks found to be executed", WorkflowStatus.COMPLETED);
            }

            WorkflowTask taskToSchedule = workflowDef.getTasks().get(0); //Nothing is running yet - so schedule the first task
            while (isTaskSkipped(taskToSchedule, workflow)) {
                taskToSchedule = workflowDef.getNextTask(taskToSchedule.getTaskReferenceName());
            }
            return getTasksToBeScheduled(workflow, taskToSchedule);
        }

        return Collections.emptyList();
    }

    /**
     * 获取下一个/一批任务节点
     * @param workflow 流程实例
     * @param task 当前任务节点
     * @return 下一个或一批任务集合
     */
    public List<Task> getNextTask(Workflow workflow, Task task) {
        final WorkflowDef workflowDef = workflow.getWorkflowDefinition();

        String taskReferenceName = task.getReferenceTaskName();
        WorkflowTask taskToSchedule = workflowDef.getNextTask(taskReferenceName);
        while (isTaskSkipped(taskToSchedule, workflow)) {
            taskToSchedule = workflowDef.getNextTask(taskToSchedule.getTaskReferenceName());
        }

        if (taskToSchedule != null) {
            return getTasksToBeScheduled(workflow, taskToSchedule);
        }
        return Collections.emptyList();
    }

    /**
     * 检查流程是否结束
     * @param workflow 流程实例
     * @return true代表流程结束
     */
    public boolean checkForWorkflowCompletion(Workflow workflow) {

        List<Task> allTasks = workflow.getTasks();
        if (allTasks.isEmpty()) {
            return false;
        }

        Map<String, Status> taskStatusMap = new HashMap<>();
        workflow.getTasks().forEach(task -> taskStatusMap.put(task.getReferenceTaskName(), task.getStatus()));

        List<WorkflowTask> workflowTasks = workflow.getWorkflowDefinition().getTasks();
        boolean allCompletedSuccessfully = workflowTasks.stream()
                .parallel()
                .allMatch(wftask -> {
                    Status status = taskStatusMap.get(wftask.getTaskReferenceName());
                    return status != null && status.isTerminal();
                });

        boolean noPendingTasks = taskStatusMap.values()
                .stream()
                .allMatch(Status::isTerminal);

        boolean noPendingSchedule = workflow.getTasks().stream()
                .parallel()
                .noneMatch(wftask -> {
                    String next = getNextTasksToBeScheduled(workflow, wftask);
                    return next != null && !taskStatusMap.containsKey(next);
                });

        return allCompletedSuccessfully && noPendingTasks && noPendingSchedule;
    }

    private String getNextTasksToBeScheduled(Workflow workflow, Task task) {
        final WorkflowDef def = workflow.getWorkflowDefinition();

        String taskReferenceName = task.getReferenceTaskName();
        WorkflowTask taskToSchedule = def.getNextTask(taskReferenceName);
        while (isTaskSkipped(taskToSchedule, workflow)) {
            taskToSchedule = def.getNextTask(taskToSchedule.getTaskReferenceName());
        }
        return taskToSchedule == null ? null : taskToSchedule.getTaskReferenceName();
    }

    public List<Task> getTasksToBeScheduled(Workflow workflow, WorkflowTask taskToSchedule) {
        Map<String, Object> input = parametersUtils.getTaskInput(taskToSchedule.getInputParameters(),
                workflow, null, null);

        List<String> tasksInWorkflow = workflow.getTasks().stream()
                .map(Task::getReferenceTaskName)
                .collect(Collectors.toList());
        String taskId = IDGenerator.generate();
        TaskMapperContext taskMapperContext = TaskMapperContext.newBuilder()
                .withWorkflowDefinition(workflow.getWorkflowDefinition())
                .withWorkflowInstance(workflow)
                .withTaskDefinition(taskToSchedule.getTaskDefinition())
                .withTaskToSchedule(taskToSchedule)
                .withTaskInput(input)
                .withTaskId(taskId)
                .withDeciderService(this)
                .build();

        List<Task> tasks = taskMappers.get(taskToSchedule.getType())
                .getMappedTasks(taskMapperContext)
                .stream()
                .filter(task -> !tasksInWorkflow.contains(task.getReferenceTaskName()))
                .collect(Collectors.toList());

        log.debug("tasks: {}", tasks);
        return tasks;
    }

    /**
     * 决策结论
     */
    public static class DeciderOutcome {
        List<Task> tasksToBeScheduled = new LinkedList<>();
        boolean isComplete;
    }

    private boolean isTaskSkipped(WorkflowTask taskToSchedule, Workflow workflow) {
        try {
            boolean isTaskSkipped = false;
            if (taskToSchedule != null) {
                Task t = workflow.getTaskByRefName(taskToSchedule.getTaskReferenceName());
                if (t == null) {
                    isTaskSkipped = false;
                } else if (t.getStatus().equals(SKIPPED)) {
                    isTaskSkipped = true;
                }
            }
            return isTaskSkipped;
        } catch (Exception e) {
            throw new TerminateWorkflowException(e.getMessage());
        }
    }

    public void updateWorkflowOutput(Workflow workflow, Task task) {
        List<Task> allTasks = workflow.getTasks();
        if (allTasks.isEmpty()) {
            return;
        }

        WorkflowDef workflowDef = workflow.getWorkflowDefinition();
        if (workflowDef.getOutputParameters() != null) {
            Map<String, Object> output = parametersUtils.getTaskInput(workflowDef.getOutputParameters(), workflow, null, null);
            workflow.setOutput(output);
        } else {
            Task last = Optional.ofNullable(task).orElse(allTasks.get(allTasks.size() - 1));
            Map<String, Object> output;
            output = last.getOutputData();
            workflow.setOutput(output);
        }
    }
}
